import re
import json
import random
import socket

from django.conf import settings
from django.core.cache import BaseCache
from django.utils.module_loading import import_string
from django.core.cache.backends.base import get_key_func

from redis.connection import DefaultParser
from redis.exceptions import ConnectionError, ResponseError, TimeoutError

_main_exceptions = (TimeoutError, ResponseError, ConnectionError, socket.timeout)


class ConnectionFactory:
    # Store connection pool by cache backend options.
    #
    # _pools is a process-global, as otherwise _pools is cleared every time
    # ConnectionFactory is instantiated, as Django creates new cache client
    # (DefaultClient) instance for every request.

    _pools = {}

    def __init__(self, options):
        pool_cls_path = options.get("CONNECTION_POOL_CLASS", "redis.connection.ConnectionPool")
        self.pool_cls = import_string(pool_cls_path)
        self.pool_cls_kwargs = options.get("CONNECTION_POOL_KWARGS", {})

        redis_client_cls_path = options.get("REDIS_CLIENT_CLASS", "redis.client.Redis")
        self.redis_client_cls = import_string(redis_client_cls_path)
        self.redis_client_cls_kwargs = options.get("REDIS_CLIENT_KWARGS", {})

        self.options = options

    def connect(self, url):
        """
        Given a basic connection parameters,
        return a new connection.
        """
        params = self.make_connection_params(url)
        key = params["url"]
        if key not in self._pools:
            self._pools[key] = self.get_connection_pool(params)
        pool = self._pools[key]

        connection = self.redis_client_cls(
            connection_pool=pool,
            **self.redis_client_cls_kwargs
        )

        return connection

    def disconnect(self, connection):
        """
        Given a not null client connection it disconnect from the Redis server.

        The default implementation uses a pool to hold connections.
        """
        connection.connection_pool.disconnect()

    def get_connection_pool(self, params):
        """
        Given a connection parameters, return a new
        connection pool for them.
        """
        cp_params = dict(params)
        cp_params.update(self.pool_cls_kwargs)
        pool = self.pool_cls.from_url(**cp_params)

        if pool.connection_kwargs.get("password", None) is None:
            pool.connection_kwargs["password"] = params.get("password", None)
            pool.reset()

        return pool

    def make_connection_params(self, url):
        """
        Given a main connection parameters, build a complete
        dict of connection parameters.
        """
        cls = self.options.get("PARSER_CLASS")
        if cls:
            parser_cls = import_string(cls)
        else:
            parser_cls = DefaultParser

        kwargs = {
            "url": url,
            "parser_class": parser_cls,
        }

        password = self.options.get("PASSWORD", None)
        if password:
            kwargs["password"] = password

        socket_timeout = self.options.get("SOCKET_TIMEOUT", None)
        if socket_timeout:
            assert isinstance(socket_timeout, (int, float)), "Socket timeout should be float or integer"
            kwargs["socket_timeout"] = socket_timeout

        socket_connect_timeout = self.options.get("SOCKET_CONNECT_TIMEOUT", None)
        if socket_connect_timeout:
            assert isinstance(socket_connect_timeout, (int, float)), "Socket connect timeout should be float or integer"
            kwargs["socket_connect_timeout"] = socket_connect_timeout

        return kwargs


# 目前仅支持一种格式的序列化器与压缩器，就不另外起文件了
class JSONSerializer:
    def __init__(self, options):
        pass

    def dumps(self, value):
        return json.dumps(value).encode()

    def loads(self, value):
        return json.loads(value.decode())


class IdentityCompressor:
    def __init__(self, options):
        self._options = options

    def compress(self, value):
        return value

    def decompress(self, value):
        return value


class ConnectionInterrupted(Exception):
    def __init__(self, connection, parent=None):
        self.connection = connection

    def __str__(self) -> str:
        error_type = type(self.__cause__).__name__
        error_msg = str(self.__cause__)
        return f"Redis {error_type}: {error_msg}"


class Client:
    def __init__(self, server, params, backend):
        self._backend = backend
        self._server = server
        self._params = params

        self.reverse_key = get_key_func(params.get("REVERSE_KEY_FUNCTION") or self.restore_key)

        if not isinstance(self._server, (list, tuple, set)):
            self._server = self._server.split(",")

        self._clients = [None] * len(self._server)
        self._options = params.get("OPTIONS", {})

        serializer_path = self._options.get("SERIALIZER")
        if serializer_path:
            serializer_cls = import_string(serializer_path)
        else:
            serializer_cls = JSONSerializer

        compressor_path = self._options.get("COMPRESSOR")
        if compressor_path:
            compressor_cls = import_string(compressor_path)
        else:
            compressor_cls = IdentityCompressor

        self._serializer = serializer_cls(options=self._options)
        self._compressor = compressor_cls(options=self._options)

        path = getattr(settings, "DJANGO_REDIS_CONNECTION_FACTORY", None)
        if path:
            redis_connection_factory = import_string(path)
        else:
            redis_connection_factory = ConnectionFactory
        self.connection_factory = redis_connection_factory(self._options)

    def __contains__(self, key):
        return self.exists(key)

    def add(self, key, value, timeout=None, version=None, client=None):
        """
        Add a value to the cache, failing if the key already exists.

        Returns ``True`` if the object was added, ``False`` if not.
        """
        return self.set(key, value, timeout, version=version, client=client, nx=True)

    def clear(self, client=None):
        """
        Flush all cache keys.
        """

        if client is None:
            client = self.get_client(write=True)

        try:
            client.flushdb()
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def close(self, **kwargs):
        close_flag = self._options.get("CLOSE_CONNECTION", getattr(settings, "DJANGO_REDIS_CLOSE_CONNECTION", False))
        if close_flag:
            num_clients = len(self._clients)
            for idx in range(num_clients):
                self.disconnect(index=idx)
            self._clients = [None] * num_clients

    def connect(self, index=0):
        """
        Given a connection index, returns a new raw redis client/connection
        instance. Index is used for replication setups and indicates that
        connection string should be used. In normal setups, index is 0.
        """
        return self.connection_factory.connect(self._server[index])

    def decode(self, value):
        """
        Decode the given value.
        """
        try:
            value = int(value)
        except (ValueError, TypeError):
            try:
                value = self._compressor.decompress(value)
            # except CompressorError:
            except Exception:
                # Handle little values, chosen to be not compressed
                pass
            value = self._serializer.loads(value)
        return value

    def decr(self, key, delta=1, version=None, client=None):
        """
        Decrease delta to value in the cache. If the key does not exist, raise a ValueError exception.
        """
        return self.incr(key=key, delta=-delta, version=version, client=client)

    def delete(self, key, version=None, prefix=None, client=None):
        """
        Remove a key from the cache.
        """
        if client is None:
            client = self.get_client(write=True)

        try:
            key = self.make_key(key, version=version, prefix=prefix)
            return client.delete(key)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def delete_many(self, keys, version=None, client=None):
        """
        Remove multiple keys at once.
        """

        if client is None:
            client = self.get_client(write=True)

        keys = [self.make_key(k, version=version) for k in keys]

        if not keys:
            return

        try:
            return client.delete(*keys)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def delete_pattern(self, pattern, version=None, prefix=None, client=None, itersize=None):
        """
        Remove all keys matching pattern.
        """

        if client is None:
            client = self.get_client(write=True)

        # pattern 合法化
        pattern = self.make_pattern(pattern, version=version, prefix=prefix)

        try:
            count = 0
            for key in client.scan_iter(match=pattern, count=itersize):
                client.delete(key)
                count += 1
            return count
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def disconnect(self, index=0, client=None):
        """
        delegates the connection factory to disconnect the client
        """
        if not client:
            client = self._clients[index]
        return self.connection_factory.disconnect(client) if client else None

    def encode(self, value):
        """
        Encode the given value.
        """
        if isinstance(value, bool) or not isinstance(value, int):
            value = self._serializer.dumps(value)
            value = self._compressor.compress(value)
            return value
        return value

    def exists(self, key, version=None, client=None):
        """
        Test if key exists.
        """

        if client is None:
            client = self.get_client(write=False)

        key = self.make_key(key, version=version)
        try:
            return client.exists(key) == 1
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def expire(self, key, timeout, version=None, client=None):
        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)

        if client.exists(key):
            client.expire(key, timeout)

    def expire_at(self, key, when, version=None, client=None):
        """
        Set an expire flag on a ``key`` to ``when``, which can be represented
        as an integer indicating unix time or a Python datetime object.
        """
        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)

        if client.exists(key):
            client.expireat(key, when)

    def get(self, key, default=None, version=None, client=None):
        """
        Retrieve a value from the cache.

        Returns decoded value if key is found, the default if not.
        """
        if client is None:
            client = self.get_client(write=False)

        key = self.make_key(key, version=version)

        try:
            value = client.get(key)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

        if value is None:
            return default

        return self.decode(value)

    def get_client(self, write=True, tried=None, show_index=False):
        """
        Method used for obtain a raw redis client.

        This function is used by almost all cache backend operations for 
        obtain a native redis client/connection instance.
        """
        if tried is None:
            tried = list()

        if tried and len(tried) < len(self._server):
            not_tried = [i for i in range(0, len(self._server)) if i not in tried]
            index = random.choice(not_tried)
        elif write or len(self._server) == 1:
            index = 0
        else:
            index = random.randint(1, len(self._server) - 1)

        if self._clients[index] is None:
            self._clients[index] = self.connect(index)

        if show_index:
            return self._clients[index], index
        else:
            return self._clients[index]

    def get_many(self, keys, version=None, client=None):
        """
        Retrieve many keys.
        """

        if client is None:
            client = self.get_client(write=False)

        if not keys:
            return dict()

        recovered_data = dict()

        map_keys = {self.make_key(k, version=version): k for k in keys}

        try:
            results = client.mget(*map_keys)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

        for key, value in zip(map_keys, results):
            if value is None:
                continue
            recovered_data[map_keys[key]] = self.decode(value)
        return recovered_data

    def incr(self, key, delta=1, version=None, client=None, ignore_key_check=False):
        """
        Add delta to value in the cache. If the key does not exist, raise a
        ValueError exception. if ignore_key_check=True then the key will be
        created and set to the delta value by default.
        """
        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)

        try:
            # if key expired after exists check, then we get key with wrong value and ttl -1.
            # use lua script for atomicity
            if not ignore_key_check:
                lua = """
                local exists = redis.call('EXISTS', KEYS[1])
                if (exists == 1) then
                    return redis.call('INCRBY', KEYS[1], ARGV[1])
                else return false end
                """
            else:
                lua = """
                return redis.call('INCRBY', KEYS[1], ARGV[1])
                """
            value = client.eval(lua, 1, key, delta)
            if value is None:
                raise ValueError("Key '%s' not found" % key)
        except ResponseError:
            # if cached value or total value is greater than 64 bit signed integer.
            # elif int is encoded. so redis sees the data as string.
            # In this situations redis will throw ResponseError

            # try to keep TTL of key
            timeout = self.ttl(key, version=version, client=client)

            # returns -2 if the key does not exist
            # means, that key have expired
            if timeout == -2:
                raise ValueError("Key '%s' not found" % key)
            value = self.get(key, version=version, client=client) + delta
            self.set(key, value, version=version, timeout=timeout, client=client)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

        return value

    def incr_version(self, key, delta=1, version=None, client=None):
        """
        Adds delta to the cache version for the supplied key. Returns the new version.
        """
        if client is None:
            client = self.get_client(write=True)

        if version is None:
            version = self._backend.version

        old_key = self.make_key(key, version)
        value = self.get(old_key, version=version, client=client)

        try:
            ttl = self.ttl(old_key, version=version, client=client)
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

        if value is None:
            raise ValueError("Key '%s' not found" % key)

        new_key = self.make_key(key, version=version + delta)

        self.set(new_key, value, timeout=ttl, client=client)
        self.delete(old_key, client=client)
        return version + delta

    def iter_keys(self, search, itersize=None, client=None, version=None):
        """
        Same as keys, but uses redis >= 2.8 cursors for make memory efficient keys iteration.
        """

        if client is None:
            client = self.get_client(write=False)

        pattern = self.make_pattern(search, version=version)
        for item in client.scan_iter(match=pattern, count=itersize):
            yield self.reverse_key(item.decode())

    def keys(self, search, version=None, client=None):
        """
        Execute KEYS command and return matched results.
        Warning: this can return huge number of results, in this case, it strongly recommended use iter_keys for it.
        """
        if client is None:
            client = self.get_client(write=False)

        pattern = self.make_pattern(search, version=version)
        try:
            return [self.reverse_key(k.decode()) for k in client.keys(pattern)]
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def lock(self, key, version=None, timeout=None, sleep=0.1, blocking_timeout=None, client=None, thread_local=True):
        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)
        return client.lock(key, timeout=timeout, sleep=sleep, blocking_timeout=blocking_timeout,
                           thread_local=thread_local)

    def make_key(self, key, version=None, prefix=None):
        if prefix is None:
            prefix = self._backend.key_prefix

        if version is None:
            version = self._backend.version

        # '%s:%s:%s' % (key_prefix, version, key)
        return self._backend.key_func(key, prefix, version)

    def make_pattern(self, pattern, version=None, prefix=None):
        # *?[这几个字符在正则中有特殊含义，替换为[*] [?] [[] 这样的字符匹配正则查找
        special_re = re.compile("([*?[])")

        if prefix is None:
            prefix = self._backend.key_prefix
        prefix = special_re.sub(r"[\1]", prefix)

        if version is None:
            version = self._backend.version
        version_str = special_re.sub(r"[\1]", str(version))

        return self._backend.key_func(pattern, prefix, version_str)

    def pexpire(self, key, timeout, version=None, client=None):
        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)

        if client.exists(key):
            client.pexpire(key, timeout)

    def pttl(self, key, version=None, client=None):
        """
        Executes PTTL redis command and return the "time-to-live" of specified key.
        If key is a non volatile key, it returns None.
        """
        if client is None:
            client = self.get_client(write=False)

        key = self.make_key(key, version=version)
        if not client.exists(key):
            return 0

        t = client.pttl(key)

        if t >= 0:
            return t
        elif t == -1:
            return None
        elif t == -2:
            return 0
        else:
            # Should never reach here
            return None

    def restore_key(self, key):
        return key.split(":", 2)[2]

    def set(self, key, value, timeout=None, version=None, client=None, nx=False, xx=False):
        """
        Persist a value to the cache, and set an optional expiration time.

        Also supports optional nx parameter. If set to True - will use redis
        setnx instead of set.
        """
        format_key = self.make_key(key, version=version)
        format_value = self.encode(value)

        if timeout is None:
            timeout = self._backend.default_timeout

        original_client = client
        tried = []
        index = 0
        while True:
            try:
                if client is None:
                    client, index = self.get_client(write=True, tried=tried, show_index=True)

                if timeout is not None:
                    # Convert to milliseconds
                    timeout = int(timeout * 1000)

                    if timeout <= 0:
                        if nx:
                            # Using negative timeouts when nx is True should
                            # not expire (in our case delete) the value if it exists.
                            # Obviously expire not existent value is noop.
                            return not self.exists(key, version=version, client=client)
                        else:
                            # redis doesn't support negative timeouts in ex flags
                            # so it seems that it's better to just delete the key
                            # than to set it and than expire in a pipeline
                            return bool(
                                self.delete(key, client=client, version=version)
                            )

                return bool(client.set(format_key, format_value, nx=nx, px=timeout, xx=xx))
            except _main_exceptions as e:
                if not original_client and len(tried) < len(self._server):
                    tried.append(index)
                    client = None
                    continue
                raise ConnectionInterrupted(connection=client) from e

    def set_many(self, data, timeout=None, version=None, client=None):
        """
        Set a bunch of values in the cache at once from a dict of key/value
        pairs. This is much more efficient than calling set() multiple times.

        If timeout is given, that timeout will be used for the key; otherwise
        the default cache timeout will be used.
        """
        if client is None:
            client = self.get_client(write=True)

        try:
            pipeline = client.pipeline()
            for key, value in data.items():
                self.set(key, value, timeout, version=version, client=pipeline)
            pipeline.execute()
        except _main_exceptions as e:
            raise ConnectionInterrupted(connection=client) from e

    def touch(self, key, timeout=None, version=None, client=None):
        """
        Sets a new expiration for a key.
        """

        if timeout is None:
            timeout = self._backend.default_timeout

        if client is None:
            client = self.get_client(write=True)

        key = self.make_key(key, version=version)
        if timeout is None:
            return bool(client.persist(key))
        else:
            # Convert to milliseconds
            timeout = int(timeout * 1000)
            return bool(client.pexpire(key, timeout))

    def ttl(self, key, version=None, client=None):
        """
        Executes TTL redis command and return the "time-to-live" of specified key.
        If key is a non volatile key, it returns None.
        """
        if client is None:
            client = self.get_client(write=False)

        key = self.make_key(key, version=version)
        if not client.exists(key):
            return 0

        t = client.ttl(key)

        if t >= 0:
            return t
        elif t == -1:
            return None
        elif t == -2:
            return 0
        else:
            # Should never reach here
            return None


class RedisCache(BaseCache):
    def __init__(self, server, params):
        # 父类初始化
        # timeout或者TIMEOUT 默认300
        # max_entries或者OPTIONS['MAX_ENTRIES'] 默认300
        # cull_frequency或者OPTIONS['CULL_FREQUENCY'] 默认3
        # KEY_PREFIX 默认""
        # VERSION 默认1
        # KEY_FUNCTION 默认django.core.cache.default_key_func
        super().__init__(params)
        self._server = server
        self._params = params

        # 获取redis处理类
        options = params.get("OPTIONS", {})
        client_cls_path = options.get("CLIENT_CLASS")
        if client_cls_path:
            self._client_cls = import_string(client_cls_path)
        else:
            self._client_cls = Client

        # 真正使用的实例变量，使用时再进行连接
        self._client = None

    @property
    def client(self):
        """
        Lazy client connection property.
        """
        if self._client is None:
            self._client = self._client_cls(self._server, self._params, self)
        return self._client

    def set(self, *args, **kwargs):
        return self.client.set(*args, **kwargs)

    def incr_version(self, *args, **kwargs):
        return self.client.incr_version(*args, **kwargs)

    def add(self, *args, **kwargs):
        return self.client.add(*args, **kwargs)

    def get(self, key, default=None, version=None, client=None):
        value = self.client.get(key, default=default, version=version, client=client)
        return value

    def delete(self, *args, **kwargs):
        """returns a boolean instead of int since django version 3.1"""
        result = self.client.delete(*args, **kwargs)
        return bool(result)

    def delete_pattern(self, *args, **kwargs):
        django_redis_scan_itersize = getattr(settings, "DJANGO_REDIS_SCAN_ITERSIZE", 10)
        kwargs["itersize"] = kwargs.get("itersize", django_redis_scan_itersize)
        return self.client.delete_pattern(*args, **kwargs)

    def delete_many(self, *args, **kwargs):
        return self.client.delete_many(*args, **kwargs)

    def clear(self):
        return self.client.clear()

    def get_many(self, *args, **kwargs):
        return self.client.get_many(*args, **kwargs)

    def set_many(self, *args, **kwargs):
        return self.client.set_many(*args, **kwargs)

    def incr(self, *args, **kwargs):
        return self.client.incr(*args, **kwargs)

    def decr(self, *args, **kwargs):
        return self.client.decr(*args, **kwargs)

    def has_key(self, *args, **kwargs):
        return self.client.exists(*args, **kwargs)

    def keys(self, *args, **kwargs):
        return self.client.keys(*args, **kwargs)

    def iter_keys(self, *args, **kwargs):
        return self.client.iter_keys(*args, **kwargs)

    def ttl(self, *args, **kwargs):
        return self.client.ttl(*args, **kwargs)

    def pttl(self, *args, **kwargs):
        return self.client.pttl(*args, **kwargs)

    def persist(self, *args, **kwargs):
        return self.client.persist(*args, **kwargs)

    def expire(self, *args, **kwargs):
        return self.client.expire(*args, **kwargs)

    def expire_at(self, *args, **kwargs):
        return self.client.expire_at(*args, **kwargs)

    def pexpire(self, *args, **kwargs):
        return self.client.pexpire(*args, **kwargs)

    def lock(self, *args, **kwargs):
        return self.client.lock(*args, **kwargs)

    def close(self, **kwargs):
        self.client.close(**kwargs)

    def touch(self, *args, **kwargs):
        return self.client.touch(*args, **kwargs)
